package com.cennavi.funcs;

import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;
import java.util.Iterator;

import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.function.Function;
import org.apache.spark.api.java.function.VoidFunction;

import scala.Tuple2;

/** * @author  作者 :allen
    * @date 创建时间：2017年9月20日 上午9:00:39 
    * @version 1.0 
    */

public class SparkStream2MySQL implements Function<JavaPairRDD<String, String>, Void> {

	private static final long serialVersionUID = 1L;

	
	public Void call(JavaPairRDD<String, String> arg0) throws Exception {
		
		if(null != arg0){
			//String topic = this.outTopic;
			final long messageSize = arg0.count();
			System.out.println("messageSize = " + messageSize);
				arg0.foreachPartition(new VoidFunction<Iterator<Tuple2<String, String>>>() {

					private static final long serialVersionUID = 1L;

					public void call(Iterator<Tuple2<String, String>> arg0) throws Exception {
						int count = 0;
						while (arg0.hasNext() && count < 1) {
							Connection conn = null;
							PreparedStatement stmt = null;
							String message = arg0.next()._2;
							try {
								String mysqlUrl = "jdbc:mysql://127.0.0.1:3306/test";
								String userName = "root";
								String password = "sorry";
								String sql = "insert into roadflowcapacity values('" + message.split(",")[0] + "','" + message.split(",")[1] + "','" + message.split(",")[2] + "','" + message.split(",")[3] + "','" + message.split(",")[4] + "','" + message.split(",")[5] + "','" + message.split(",")[6] + "','" + message.split(",")[7] + "','" + message.split(",")[8] + "','" + message.split(",")[9] + "','" + message.split(",")[10] + "','" + message.split(",")[11] + "','" + message.split(",")[12] + "')";
								conn = DriverManager.getConnection(mysqlUrl, userName, password);
								stmt = conn.prepareStatement(sql);
								stmt.executeUpdate();
							} catch (Exception e) {
								e.printStackTrace();
							} finally {
								if (stmt != null) {  
						            stmt.close();
						          }  
						          if (conn != null) {  
						            conn.close(); 
						          }
							}
							count ++;
						}
						
					}});
		}
		return null;
	}
	
//	public static void main(String[] args) throws ParseException {
//		String date1 = "2017-09-25 7:00";
//		String date2 = "2020-10-08 7:00";
//		long deltaDays = (new SimpleDateFormat("yyyy-MM-dd HH:mm").parse(date2).getTime()
//				- new SimpleDateFormat("yyyy-MM-dd HH:mm").parse(date1).getTime()) / (1000 * 60 * 60 * 24);
//		System.out.println(deltaDays);
		
//		List<String> test = new ArrayList<String>();
//		test.add("aaa");
//		test.add("bbb");
//		test.add("ccc");
//		Iterator it = test.iterator();
//		int count = 0;
//		while(it.hasNext()){
//			count ++;
//			it.next();
//			if(!it.hasNext()){
//				System.out.println(count);
//			}
//		}
//		StringBuilder sb = new StringBuilder();
//		sb.append("first");
//		sb.setLength(0);
//		System.out.println(sb.toString());
//		sb.append("second");
//		System.out.println(sb.toString());
//	}
}
